package com.wtz.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.List;
import java.util.Properties;

/**
 * Date: 2020/12/30
 *
 * @author tiezhu
 * Company dtstack
 */
public class KafkaConsumerWithPartition {
    private static final String BROKER_LIST = "kudu:9092";
    private static final String GROUP_ID = "*";

    public static void main(String[] args) {
        KafkaConsumer<String, String> kafkaConsumer = buildKafkaConsumer();
        List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor(args[0]);
        System.out.println(partitionInfos.toString());
        while(true) {

        }
    }

    private static KafkaConsumer<String, String> buildKafkaConsumer() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        return new KafkaConsumer<>(properties);
    }
}
